package com.sshine.solon.rabbitmq.impl;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.sshine.solon.rabbitmq.RabbitAdmin;
import com.sshine.solon.rabbitmq.builder.MessageBuilder;
import com.sshine.solon.rabbitmq.core.Message;
import com.sshine.solon.rabbitmq.core.MessageProperties;
import com.sshine.solon.rabbitmq.service.RabbitMessageHandler;
import com.sshine.solon.rabbitmq.utils.ExpirationUtils;
import org.noear.solon.Utils;
import org.noear.solon.core.event.EventBus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;

/**
 * @author noear
 * @since 1.3
 */
public class RabbitConsumeHandler extends DefaultConsumer {
    static Logger log = LoggerFactory.getLogger(RabbitConsumeHandler.class);
    private RabbitMessageHandler handler;
    private RabbitAdmin rabbitAdmin;
    private String queue;

    public RabbitConsumeHandler(Channel channel, RabbitMessageHandler handler, RabbitAdmin rabbitAdmin) {
        super(channel);
        this.handler = handler;
        this.rabbitAdmin = rabbitAdmin;
    }

    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        try {
            MessageProperties messageProperties = this.rabbitAdmin.getMessagePropertiesConverter().toMessageProperties(properties, envelope,properties.getContentEncoding());
            Message message = MessageBuilder.withBody(body).andProperties(messageProperties).build();

            boolean isOk = onReceive(message); //吃掉异常，方便下面的动作

            if (isOk == false) {
                int times = (int) messageProperties.getHeader(RabbitAdmin.RETRY_TIMES);
                messageProperties.setHeader(RabbitAdmin.RETRY_TIMES,times + 1);
                try {
                    isOk = rabbitAdmin.convertAndSend(envelope.getExchange(), envelope.getRoutingKey(),message, ExpirationUtils.getExpiration(times));
                } catch (Throwable ex) {
                    getChannel().basicNack(envelope.getDeliveryTag(), false, true);
                    isOk = true;
                }
            }

            if (isOk) {
                getChannel().basicAck(envelope.getDeliveryTag(), false);
            }

        } catch (Throwable ex) {
            ex = Utils.throwableUnwrap(ex);

            EventBus.push(ex);

            if (ex instanceof RuntimeException) {
                throw (RuntimeException) ex;
            } else if (ex instanceof IOException) {
                throw (IOException) ex;
            } else {
                throw new RuntimeException(ex);
            }
        }
    }

    private boolean onReceive(Message message) throws Throwable {
        try {
            return handler.handle(message);
        } catch (Throwable e) {
            EventBus.push(e);
            return false;
        }
    }
}